package cn.xuexiyuan.flinkstudy.watermark;

import cn.xuexiyuan.flinkstudy.entity.Order;
import cn.xuexiyuan.flinkstudy.test.DataFactory;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @Description: 基与事件时间的窗口计算 + watermark 解决一定程度数据乱序/延迟到达的问题
 *
 * @Author 左龙龙
 * @Date 21-3-25
 * @Version 1.0
 **/
public class WatermarkDemo02 {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<Order> orderDS = env.addSource(new SourceFunction<Order>() {
            private boolean running = true;
            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                while(running){
                    Order order = DataFactory.createRandomOrder(2, 5);
                    System.out.println(JSON.toJSONString(order));
                    ctx.collect(order);
                    Thread.sleep(3000);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        });

        // 2.transformation
//        SingleOutputStreamOperator<ChannelOrder> orderDSWithWatermark = orderDS.assignTimestampsAndWatermarks(
//                // 设置 maxOutOfOrderness 最大无序度/最大允许的延迟/乱序时间
//                WatermarkStrategy.<ChannelOrder>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//                        // 指定事件时间列
//                        .withTimestampAssigner((order, timestamp) -> order.getOrder_time().getTime())
//        );

        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
        SingleOutputStreamOperator<Order> orderDSWithWatermark = orderDS.assignTimestampsAndWatermarks(
                new WatermarkStrategy<Order>() {
                    @Override
                    public WatermarkGenerator<Order> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                        return new WatermarkGenerator<Order>() {
                            private int user_id;
                            private long order_time;
                            private final long outOfOrdernessMillis = 3000;
                            private long maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;


                            @Override
                            public void onEvent(Order o, long eventTimestamp, WatermarkOutput watermarkOutput) {
                                user_id = o.getUser_id();
                                order_time = o.getOrder_time();
                                maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
                                System.out.println("[onEvent]eventTimestamp:" + sdf.format(eventTimestamp));
                            }

                            @Override
                            public void onPeriodicEmit(WatermarkOutput output) {
                                // watermark = 当前最大事件时间 - 最大允许延迟时间
                                Watermark watermark = new Watermark(maxTimestamp - outOfOrdernessMillis - 1);

                                System.out.println("[onPeriodicEmit]key: " + user_id
                                        + ",系统时间:" + sdf.format(new Date())
                                        + ",事件时间：" + sdf.format(order_time)
                                        + ",水印时间：" + sdf.format(watermark.getTimestamp()));
                                output.emitWatermark(watermark);
                            }
                        };
                    }
                }.withTimestampAssigner((event, timestamp) -> event.getOrder_time())
        );


        // 学习测试使用如下代码对数据进行更详细的输出, 如输出窗口触发时各个窗口中数据的事件时间和watermark 时间
        SingleOutputStreamOperator<String> result = orderDSWithWatermark
                .keyBy(Order::getUser_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // 把 apply 中的函数应用在窗口中的数据上
                .apply(new WindowFunction<Order, String, Integer, TimeWindow>() {
                    @Override
                    public void apply(Integer key, TimeWindow window, Iterable<Order> input, Collector<String> out) throws Exception {
                        // 准备一个集合用来存放属于该窗口的数据的事件时间
                        List<String> eventTimeList = new ArrayList();
                        for(Order order : input){
                            eventTimeList.add(sdf.format(order.getOrder_time()));
                        }
                        String str = String.format("key: %s ,系统时间:%s ,窗口时间范围[%s~%s), 属于该窗口的事件时间： %s", key, sdf.format(new Date()),  sdf.format(window.getStart()), sdf.format(window.getEnd()), eventTimeList);
                        out.collect(str);
                    }
                });


        // 3.sink
        result.print();

        // 4.excute
        env.execute();

    }

}
